package com.kiss.rpc;

import com.kiss.rpc.channelhandler.handler.MethodCallHandler;
import com.kiss.rpc.channelhandler.handler.RpcRequestDecoder;
import com.kiss.rpc.channelhandler.handler.RpcResponseEncoder;
import com.kiss.rpc.config.Configuration;
import com.kiss.rpc.core.HeartbeatDetector;
import com.kiss.rpc.discovery.RegistryConfig;
import com.kiss.rpc.transport.message.RpcRequest;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.ResourceLeakDetector;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class RpcBootstrap {
    private static final RpcBootstrap rpcBootstrap = new RpcBootstrap();
    private static Configuration config;

    public static final ThreadLocal<RpcRequest> REQUEST_THREAD_LOCAL = new ThreadLocal<>();
    public static final Map<InetSocketAddress, Channel> CHANNEL_CACHE = new ConcurrentHashMap<>(16);
    public static final TreeMap<Long, Channel> ANSWER_TIME_CHANNEL_CACHE = new TreeMap<>();

    public static final Map<String, ServiceConfig<?>> SERVICE_LIST = new ConcurrentHashMap<>(16);
    public static final Map<Long, CompletableFuture<Object>> PENDING_REQUEST = new ConcurrentHashMap<>(128);

    private static final AtomicLong count = new AtomicLong(0);

    private RpcBootstrap() {
        config = new Configuration();
    }

    public static RpcBootstrap getInstance() {
        return rpcBootstrap;
    }

    public static void addOne() {
        count.incrementAndGet();
    }

    public RpcBootstrap application(String appName) {
        config.setAppName(appName);
        return this;
    }

    public RpcBootstrap registry(RegistryConfig registryConfig) {
        config.setRegistryConfig(registryConfig);
        return this;
    }

    public RpcBootstrap publish(ServiceConfig<?> serviceConfig) {
        config.getRegistryConfig().getRegistry().register(serviceConfig);
        SERVICE_LIST.put(serviceConfig.getInterface().getName(), serviceConfig);
        return this;
    }

    public RpcBootstrap publish(List<ServiceConfig<?>> serviceConfigs) {
        for (ServiceConfig<?> serviceConfig : serviceConfigs) {
            publish(serviceConfig);
        }
        return this;
    }

    public RpcBootstrap reference(ReferenceConfig<?> referenceConfig) {
        HeartbeatDetector.detectHeartbeat(referenceConfig.getInterfaceRef().getName());

        referenceConfig.setRegistry(config.getRegistryConfig().getRegistry());
        referenceConfig.setGroup(config.getGroup());
        return this;
    }

    public Configuration getConfig() {
        return config;
    }

    public void start() {
        Thread thread = new Thread(() ->
                new Timer().scheduleAtFixedRate(new TimerTask() {
                    @Override
                    public void run() {
                        System.out.println("QPS is " + count.get());
                        System.out.flush();
                        count.set(0);
                    }
                }, 0, 1000),
                "rpc-QPS-Thread");
        thread.setDaemon(true);
        thread.start();

        try (
                EventLoopGroup boss = new NioEventLoopGroup(2);
                EventLoopGroup worker = new NioEventLoopGroup(8);
                ){
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap = serverBootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new LoggingHandler())
                                    .addLast(new RpcRequestDecoder())
                                    .addLast(new MethodCallHandler())
                                    .addLast(new RpcResponseEncoder());
                        }
                    });
            ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
            ChannelFuture channelFuture = serverBootstrap.bind(config.getPort()).sync();

            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        RpcBootstrap rpcBootstrap = RpcBootstrap.getInstance();
        rpcBootstrap.start();
    }
}
